//! 单线程命令分发器
//! 
//! 参考 Redis 官方设计，采用单线程单队列架构，
//! 所有命令统一处理，保证操作原子性和数据一致性。

use crate::error::Result;
use crate::storage::{Command, HighPerformanceStorage, HighPerformanceStorageStats};
use crate::event::{
    Event, EventPayload, DynamicEventHandler, EventResult, 
    CrossbeamEventDispatcher, CrossbeamDispatcherConfig, CrossbeamEventSender
};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Instant;
use std::cell::UnsafeCell;
use tracing::{debug, info, warn/*, error*/};

/// 命令事件负载
#[derive(Debug)]
pub struct CommandEvent {
    /// 待执行的命令
    pub command: Box<dyn Command>,
    /// 创建时间戳
    pub timestamp: Instant,
}

impl CommandEvent {
    pub fn new(command: Box<dyn Command>) -> Self {
        Self {
            command,
            timestamp: Instant::now(),
        }
    }
}

impl EventPayload for CommandEvent {
    fn payload_type(&self) -> &'static str {
        "redis_command"
    }

    fn serialize(&self) -> std::result::Result<Vec<u8>, Box<dyn std::error::Error + Send + Sync>> {
        let args = self.command.to_args();
        let mut lines = vec![self.command.name().to_string()];
        lines.extend(args);
        let serialized = lines.join("\n");
        Ok(serialized.into_bytes())
    }

    fn size(&self) -> usize {
        self.command.name().len() + self.command.to_args().iter().map(|s| s.len()).sum::<usize>()
    }
}

/// 单线程 Redis 命令处理器（高性能版本）
pub struct SingleThreadCommandHandler {
    /// 高性能存储引擎 - 使用 UnsafeCell 允许内部可变性
    storage: UnsafeCell<HighPerformanceStorage>,
    /// 处理器名称
    name: String,
    /// 统计信息
    processed_count: AtomicU64,
    success_count: AtomicU64,
    error_count: AtomicU64,
    total_processing_time: AtomicU64,
}

// 在单线程环境中，这是安全的
unsafe impl Sync for SingleThreadCommandHandler {}
unsafe impl Send for SingleThreadCommandHandler {}

impl SingleThreadCommandHandler {
    pub fn new(name: String) -> Self {
        info!("Creating single-thread command handler with high-performance storage: {}", name);
        Self {
            storage: UnsafeCell::new(HighPerformanceStorage::new()),
            name,
            processed_count: AtomicU64::new(0),
            success_count: AtomicU64::new(0),
            error_count: AtomicU64::new(0),
            total_processing_time: AtomicU64::new(0),
        }
    }

    /// 获取高性能存储引擎的引用
    pub fn storage(&self) -> &HighPerformanceStorage {
        unsafe { &*self.storage.get() }
    }

    /// 获取高性能存储引擎的可变引用
    #[allow(clippy::mut_from_ref)]
    fn storage_mut(&self) -> &mut HighPerformanceStorage {
        unsafe { &mut *self.storage.get() }
    }

    /// 获取统计信息
    pub fn get_stats(&self) -> CommandHandlerStats {
        CommandHandlerStats {
            name: self.name.clone(),
            processed_count: self.processed_count.load(Ordering::Relaxed),
            success_count: self.success_count.load(Ordering::Relaxed),
            error_count: self.error_count.load(Ordering::Relaxed),
            avg_processing_time_us: {
                let total_time = self.total_processing_time.load(Ordering::Relaxed);
                let processed = self.processed_count.load(Ordering::Relaxed);
                if processed > 0 {
                    total_time / processed
                } else {
                    0
                }
            },
        }
    }

    /// 执行 Redis 命令
    fn execute_command(&self, event: &Event<Box<dyn EventPayload>>) -> EventResult {
        let start_time = Instant::now();
        self.processed_count.fetch_add(1, Ordering::Relaxed);

        // 解析命令数据
        let serialized = match event.payload.serialize() {
            Ok(data) => data,
            Err(e) => {
                self.error_count.fetch_add(1, Ordering::Relaxed);
                return EventResult::failed(format!("Failed to serialize command: {e}"));
            }
        };

        let content = match String::from_utf8(serialized) {
            Ok(s) => s,
            Err(e) => {
                self.error_count.fetch_add(1, Ordering::Relaxed);
                return EventResult::failed(format!("Invalid UTF-8 data: {e}"));
            }
        };

        let lines: Vec<&str> = content.lines().collect();
        if lines.is_empty() {
            self.error_count.fetch_add(1, Ordering::Relaxed);
            return EventResult::failed("Empty command data");
        }

        let command_name = lines[0];
        let args: Vec<String> = if lines.len() > 1 {
            lines[1..].iter().map(|s| s.to_string()).collect()
        } else {
            Vec::new()
        };

        // 执行具体的 Redis 命令
        let result = self.execute_redis_command(command_name, args);
        
        let processing_time = start_time.elapsed().as_micros() as u64;
        self.total_processing_time.fetch_add(processing_time, Ordering::Relaxed);

        match result {
            Ok(_) => {
                debug!("Handler {} executed command {} in {}μs", 
                       self.name, command_name, processing_time);
                self.success_count.fetch_add(1, Ordering::Relaxed);
                EventResult::Success
            }
            Err(e) => {
                warn!("Handler {} failed to execute command {}: {}", 
                      self.name, command_name, e);
                self.error_count.fetch_add(1, Ordering::Relaxed);
                EventResult::failed(format!("Command execution failed: {e}"))
            }
        }
    }

    /// 执行具体的 Redis 命令
    fn execute_redis_command(&self, command: &str, args: Vec<String>) -> Result<()> {
        let storage = self.storage_mut();
        match command.to_uppercase().as_str() {
            // String 操作
            "SET" => {
                if args.len() != 2 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "SET requires exactly 2 arguments".to_string(),
                    });
                }
                storage.set(args[0].clone(), args[1].clone())
            }
            "GET" => {
                if args.len() != 1 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "GET requires exactly 1 argument".to_string(),
                    });
                }
                storage.get(&args[0]).map(|_| ())
            }
            "INCR" => {
                if args.len() != 1 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "INCR requires exactly 1 argument".to_string(),
                    });
                }
                storage.incr(&args[0]).map(|_| ())
            }
            "INCRBY" => {
                if args.len() != 2 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "INCRBY requires exactly 2 arguments".to_string(),
                    });
                }
                let amount: i64 = args[1].parse().map_err(|_| crate::error::StorageError::InvalidArgument {
                    message: "Invalid number format for INCRBY".to_string(),
                })?;
                storage.incr_by(&args[0], amount).map(|_| ())
            }

            // Hash 操作
            "HSET" => {
                if args.len() != 3 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "HSET requires exactly 3 arguments".to_string(),
                    });
                }
                storage.hset(&args[0], args[1].clone(), args[2].clone()).map(|_| ())
            }
            "HGET" => {
                if args.len() != 2 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "HGET requires exactly 2 arguments".to_string(),
                    });
                }
                storage.hget(&args[0], &args[1]).map(|_| ())
            }
            "HDEL" => {
                if args.len() != 2 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "HDEL requires exactly 2 arguments".to_string(),
                    });
                }
                storage.hdel(&args[0], &args[1]).map(|_| ())
            }

            // 通用操作
            "DEL" => {
                if args.len() != 1 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "DEL requires exactly 1 argument".to_string(),
                    });
                }
                storage.del(&args[0]);
                Ok(())
            }
            "EXISTS" => {
                if args.len() != 1 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "EXISTS requires exactly 1 argument".to_string(),
                    });
                }
                storage.exists(&args[0]);
                Ok(())
            }
            "EXPIRE" => {
                if args.len() != 2 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "EXPIRE requires exactly 2 arguments".to_string(),
                    });
                }
                let seconds: u64 = args[1].parse().map_err(|_| crate::error::StorageError::InvalidArgument {
                    message: "Invalid number format for EXPIRE".to_string(),
                })?;
                storage.expire(&args[0], seconds);
                Ok(())
            }
            "TTL" => {
                if args.len() != 1 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "TTL requires exactly 1 argument".to_string(),
                    });
                }
                storage.ttl(&args[0]);
                Ok(())
            }

            _ => {
                Err(crate::error::StorageError::InvalidArgument {
                    message: format!("Unknown command: {command}"),
                })
            }
        }
    }
}

impl DynamicEventHandler for SingleThreadCommandHandler {
    fn handle_dynamic(&self, event: &Event<Box<dyn EventPayload>>) -> EventResult {
        // 由于需要修改存储状态，我们需要使用内部可变性
        // 这在单线程环境中是安全的
        let handler_ptr = self as *const Self as *mut Self;
        unsafe {
            (*handler_ptr).execute_command(event)
        }
    }

    fn name(&self) -> &str {
        &self.name
    }

    fn can_handle(&self, event_type: &str) -> bool {
        event_type == "redis_command"
    }
}

/// 单线程命令分发系统（高性能版本）
pub struct SingleThreadCommandDispatcher {
    /// 高性能事件分发器 - 使用crossbeam无锁通道
    dispatcher: CrossbeamEventDispatcher,
    /// 事件发送器
    sender: CrossbeamEventSender,
    /// 命令处理器（公开以便协议层访问）
    pub handler: Arc<SingleThreadCommandHandler>,
    /// 运行状态
    running: bool,
    /// 统计信息
    total_commands: AtomicU64,
}

impl SingleThreadCommandDispatcher {
    /// 创建新的单线程命令分发系统（高性能版本）
    pub fn new() -> Self {
        info!("创建高性能单线程命令分发器");
        
        // 配置高性能分发器：优先使用单线程避免竞争
        let config = CrossbeamDispatcherConfig {
            worker_count: 1, // 单线程模式，避免竞争
            enable_stats: true,
            worker_name_prefix: "single-thread".to_string(),
        };
        let mut dispatcher = CrossbeamEventDispatcher::new(config);
        
        // 创建单线程命令处理器
        let handler = Arc::new(SingleThreadCommandHandler::new("HighPerformanceHandler".to_string()));
        
        // 注册处理器
        dispatcher.register_handler("redis_command", handler.clone());
        
        // 获取发送器
        let sender = dispatcher.get_sender();

        Self {
            dispatcher,
            sender,
            handler,
            running: false,
            total_commands: AtomicU64::new(0),
        }
    }

    /// 启动分发器
    pub fn start(&mut self) {
        if self.running {
            warn!("Command dispatcher is already running");
            return;
        }

        info!("Starting single-thread command dispatcher");
        self.dispatcher.start();
        self.running = true;
        info!("Command dispatcher started successfully");
    }

    /// 停止分发器
    pub fn stop(&mut self) {
        if !self.running {
            warn!("Command dispatcher is not running");
            return;
        }

        info!("Stopping single-thread command dispatcher");
        self.dispatcher.stop();
        self.running = false;
        info!("Command dispatcher stopped successfully");
    }

    /// 分发命令（使用高性能分发器）
    pub fn dispatch_command(&self, command: Box<dyn Command>) -> Result<()> {
        if !self.running {
            return Err(crate::error::StorageError::InvalidOperation {
                message: "Command dispatcher is not running".to_string(),
            });
        }

        let command_event = CommandEvent::new(command);
        let event = Event::new(
            "redis_command",
            Box::new(command_event) as Box<dyn EventPayload>
        );

        match self.sender.publish(event) {
            Ok(()) => {
                self.total_commands.fetch_add(1, Ordering::Relaxed);
                Ok(())
            }
            Err(_) => Err(crate::error::StorageError::InvalidOperation {
                message: "Failed to dispatch command".to_string(),
            })
        }
    }

    /// 获取统计信息
    pub fn get_stats(&self) -> SingleThreadDispatcherStats {
        let handler_stats = self.handler.get_stats();
        let storage_stats = self.handler.storage().get_stats();

        SingleThreadDispatcherStats {
            total_dispatched: self.total_commands.load(Ordering::Relaxed),
            handler_stats,
            storage_stats,
            is_running: self.running,
        }
    }

    /// 检查是否正在运行
    pub fn is_running(&self) -> bool {
        self.running
    }

    /// 获取高性能存储统计信息
    pub fn get_storage_stats(&self) -> HighPerformanceStorageStats {
        self.handler.storage().get_stats()
    }
    
    /// 同步执行单个命令并返回结果（用于协议层）
    pub fn execute_single_command_sync(&self, command_name: &str, args: Vec<String>) -> crate::error::Result<crate::storage::CommandResult> {
        use crate::storage::CommandResult;
        
        // 直接调用处理器的私有方法
        let storage = unsafe { &mut *self.handler.storage.get() };
        
        // 根据命令类型执行对应操作
        match command_name.to_uppercase().as_str() {
            "PING" => Ok(CommandResult::String("PONG".to_string())),
            "SET" => {
                if args.len() != 2 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "SET requires exactly 2 arguments".to_string(),
                    });
                }
                storage.set(args[0].clone(), args[1].clone())?;
                Ok(CommandResult::Ok)
            }
            "GET" => {
                if args.len() != 1 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "GET requires exactly 1 argument".to_string(),
                    });
                }
                match storage.get(&args[0])? {
                    Some(value) => Ok(CommandResult::String(value)),
                    None => Ok(CommandResult::Null),
                }
            }
            "EXISTS" => {
                if args.len() != 1 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "EXISTS requires exactly 1 argument".to_string(),
                    });
                }
                let exists = storage.exists(&args[0]);
                Ok(CommandResult::Integer(if exists { 1 } else { 0 }))
            }
            "DEL" => {
                if args.len() != 1 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "DEL requires exactly 1 argument".to_string(),
                    });
                }
                let deleted = storage.del(&args[0]);
                Ok(CommandResult::Integer(if deleted { 1 } else { 0 }))
            }
            "INCR" => {
                if args.len() != 1 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "INCR requires exactly 1 argument".to_string(),
                    });
                }
                let result = storage.incr(&args[0])?;
                Ok(CommandResult::Integer(result))
            }
            "INCRBY" => {
                if args.len() != 2 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "INCRBY requires exactly 2 arguments".to_string(),
                    });
                }
                let amount: i64 = args[1].parse().map_err(|_| crate::error::StorageError::InvalidArgument {
                    message: "Invalid number format".to_string(),
                })?;
                let result = storage.incr_by(&args[0], amount)?;
                Ok(CommandResult::Integer(result))
            }
            "DECR" => {
                if args.len() != 1 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "DECR requires exactly 1 argument".to_string(),
                    });
                }
                let result = storage.incr_by(&args[0], -1)?;
                Ok(CommandResult::Integer(result))
            }
            "DECRBY" => {
                if args.len() != 2 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "DECRBY requires exactly 2 arguments".to_string(),
                    });
                }
                let amount: i64 = args[1].parse().map_err(|_| crate::error::StorageError::InvalidArgument {
                    message: "value is not an integer or out of range".to_string(),
                })?;
                let result = storage.incr_by(&args[0], -amount)?;
                Ok(CommandResult::Integer(result))
            }
            "HSET" => {
                if args.len() != 3 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "HSET requires exactly 3 arguments".to_string(),
                    });
                }
                let is_new = storage.hset(&args[0], args[1].clone(), args[2].clone())?;
                Ok(CommandResult::Integer(if is_new { 1 } else { 0 }))
            }
            "HGET" => {
                if args.len() != 2 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "HGET requires exactly 2 arguments".to_string(),
                    });
                }
                match storage.hget(&args[0], &args[1])? {
                    Some(value) => Ok(CommandResult::String(value)),
                    None => Ok(CommandResult::Null),
                }
            }
            "HDEL" => {
                if args.len() != 2 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "HDEL requires exactly 2 arguments".to_string(),
                    });
                }
                let deleted = storage.hdel(&args[0], &args[1])?;
                Ok(CommandResult::Integer(if deleted { 1 } else { 0 }))
            }
            "CLIENT" => {
                // CLIENT 命令的简单处理
                if args.is_empty() {
                    return Ok(CommandResult::Error("CLIENT requires subcommand".to_string()));
                }
                match args[0].to_uppercase().as_str() {
                    "SETINFO" => Ok(CommandResult::Ok),
                    "GETNAME" => Ok(CommandResult::Null),
                    "ID" => Ok(CommandResult::Integer(1)),
                    "LIST" => Ok(CommandResult::String("".to_string())),
                    _ => Ok(CommandResult::Error(format!("Unknown CLIENT subcommand: {}", args[0])))
                }
            }
            "TYPE" => {
                if args.len() != 1 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "TYPE requires exactly 1 argument".to_string(),
                    });
                }
                match storage.get_type(&args[0]) {
                    Some(type_name) => Ok(CommandResult::String(type_name)),
                    None => Ok(CommandResult::String("none".to_string())),
                }
            }
            "EXPIRE" => {
                if args.len() != 2 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "EXPIRE requires exactly 2 arguments".to_string(),
                    });
                }
                let seconds: u64 = args[1].parse().map_err(|_| crate::error::StorageError::InvalidArgument {
                    message: "Invalid number format".to_string(),
                })?;
                let result = storage.expire(&args[0], seconds);
                Ok(CommandResult::Integer(if result { 1 } else { 0 }))
            }
            "TTL" => {
                if args.len() != 1 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "TTL requires exactly 1 argument".to_string(),
                    });
                }
                let ttl = storage.ttl(&args[0]);
                Ok(CommandResult::Integer(ttl))
            }
            "PERSIST" => {
                if args.len() != 1 {
                    return Err(crate::error::StorageError::InvalidArgument {
                        message: "PERSIST requires exactly 1 argument".to_string(),
                    });
                }
                let result = storage.persist(&args[0]);
                Ok(CommandResult::Integer(if result { 1 } else { 0 }))
            }
            "FLUSHALL" => {
                storage.clear();
                Ok(CommandResult::Ok)
            }
            _ => {
                Ok(CommandResult::Error(format!("Unknown command: {command_name}")))
            }
        }
    }
}

impl Drop for SingleThreadCommandDispatcher {
    fn drop(&mut self) {
        if self.running {
            self.stop();
        }
    }
}

/// 命令处理器统计信息
#[derive(Debug, Clone)]
pub struct CommandHandlerStats {
    pub name: String,
    pub processed_count: u64,
    pub success_count: u64,
    pub error_count: u64,
    pub avg_processing_time_us: u64,
}

impl CommandHandlerStats {
    /// 获取成功率
    pub fn success_rate(&self) -> f64 {
        if self.processed_count == 0 {
            1.0
        } else {
            self.success_count as f64 / self.processed_count as f64
        }
    }
}

/// 单线程分发器统计信息
#[derive(Debug, Clone)]
pub struct SingleThreadDispatcherStats {
    pub total_dispatched: u64,
    pub handler_stats: CommandHandlerStats,
    pub storage_stats: HighPerformanceStorageStats,
    pub is_running: bool,
}

impl SingleThreadDispatcherStats {
    /// 获取成功率
    pub fn success_rate(&self) -> f64 {
        let total = self.handler_stats.processed_count;
        if total == 0 {
            1.0
        } else {
            self.handler_stats.success_count as f64 / total as f64
        }
    }

    /// 获取错误率
    pub fn error_rate(&self) -> f64 {
        let total = self.handler_stats.processed_count;
        if total == 0 {
            0.0
        } else {
            self.handler_stats.error_count as f64 / total as f64
        }
    }
}

impl Default for SingleThreadCommandDispatcher {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::storage::CommandFactory;

    #[test]
    fn test_single_thread_dispatcher_lifecycle() {
        let mut dispatcher = SingleThreadCommandDispatcher::new();
        
        // 初始状态
        assert!(!dispatcher.is_running());
        
        // 启动
        dispatcher.start();
        assert!(dispatcher.is_running());
        
        // 停止
        dispatcher.stop();
        assert!(!dispatcher.is_running());
    }

    #[test]
    fn test_command_dispatch() {
        let mut dispatcher = SingleThreadCommandDispatcher::new();
        dispatcher.start();

        // 测试 SET 命令
        let set_cmd = CommandFactory::create("SET".to_string(), vec![
            "test_key".to_string(),
            "test_value".to_string()
        ]).expect("Failed to create SET command");
        
        let result = dispatcher.dispatch_command(set_cmd);
        assert!(result.is_ok());

        // 测试 GET 命令
        let get_cmd = CommandFactory::create("GET".to_string(), vec![
            "test_key".to_string()
        ]).expect("Failed to create GET command");
        
        let result = dispatcher.dispatch_command(get_cmd);
        assert!(result.is_ok());

        // 等待处理完成
        std::thread::sleep(std::time::Duration::from_millis(100));

        let stats = dispatcher.get_stats();
        assert_eq!(stats.total_dispatched, 2);
        assert!(stats.handler_stats.processed_count >= 2);

        dispatcher.stop();
    }

    #[test]
    fn test_performance() {
        let mut dispatcher = SingleThreadCommandDispatcher::new();
        dispatcher.start();

        let start_time = Instant::now();
        const COMMAND_COUNT: u32 = 1000;

        // 执行大量命令
        for i in 0..COMMAND_COUNT {
            let set_cmd = CommandFactory::create("SET".to_string(), vec![
                format!("key_{}", i),
                format!("value_{}", i)
            ]).expect("Failed to create SET command");
            
            dispatcher.dispatch_command(set_cmd).unwrap();
        }

        // 等待处理完成
        std::thread::sleep(std::time::Duration::from_millis(500));
        
        let elapsed = start_time.elapsed();
        let _ops_per_sec = COMMAND_COUNT as f64 / elapsed.as_secs_f64();
        
        // 单线程分发器性能测试统计

        let stats = dispatcher.get_stats();
        assert_eq!(stats.total_dispatched, COMMAND_COUNT as u64);
        assert!(stats.handler_stats.success_rate() > 0.95); // 至少 95% 成功率

        dispatcher.stop();
    }
}